package com.student.xujiang.myallstudent.rxjava.rxjava2;

import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

/**
 * Created by xujiang on 2017/3/21.
 */

public class RxUtils2 {

    private static final String TAG = RxUtils2.class.getSimpleName();

    //创建一个Flowable对象很简单，直接调用Flowable.create即可。
    public static void createFlowable() {
        Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("hello RxJava 2");
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

        Subscriber subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                /**
                 * 需要注意的是，在onSubscribe中，我们需要调用request去请求资源，参数就是要请求的数量，
                 * 一般如果不限制请求数量，可以写成Long.MAX_VALUE。
                 * 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用。
                 * */
                Log.i(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete");
            }
        };

        //为了让”发射器”和”接收器”工作起来，我们还需要把他们组装在一起。
        flowable.subscribe(subscriber);

    }

    public static void just() {

//        Flowable<String> flowable = Flowable.just("hello RxJava 2");
//
//        Consumer consumer = new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Exception {
//                Log.i(TAG,s);
//            }
//        };
//
//        flowable.subscribe(consumer);
//    }

        //or
        Flowable.just("hello Rxjava2")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        Log.i(TAG, s);
                    }
                });
    }


    public static void map(){
        Flowable.just("map")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {
                        return s+"-ittianyn";
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.i(TAG,s);
            }
        });
    }

    public static void map2(){
        Flowable.just("map2").map(new Function<String, Integer>() {
            @Override
            public Integer apply(@NonNull String s) throws Exception {
                return s.hashCode();
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return integer.toString();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.i(TAG,s);
            }
        });
    }

}
